/******************************************************************************
 * Copyright 2024 zhangcong. All Rights Reserved.
 *****************************************************************************/
#include "business/recordbusiness/storetask.h"
#include "common/common_log.h"
#include "common/common_time_utils.h"
#include "zmq.h"
#include <sstream>
#include <unistd.h>

#define PLC_CFG_FILE "/app/conf/core_info"
#define DATA_ENABLE_CONFIG "/app/bin/zmq_data_enable.json"

static int32_t searchFunc(std::string filename, std::string data,
                          std::string &result) {
  std::string cmd = "grep " + data + " " + filename + "  | sed 's/^.*= //g'";
  FILE *fstream = nullptr;

  char buf[256] = {0};
  if (nullptr == (fstream = popen(cmd.c_str(), "r"))) {
    Common_Log_e("open cmd fail");
    return -1;
  }

  if (nullptr == fgets(buf, sizeof(buf), fstream)) {
    Common_Log_e("fgets buff fail");
    return -1;
  }
  result = buf;
  result = result.substr(0, result.length() - 1);

  Common_Log_e("data:%s, result:%s, leng:%d\n", data.c_str(), result.c_str(),
               result.length());

  return 0;
}

StoreTask::StoreTask(const StoreTaskInfo &info) : m_task_info(info) {
  m_SensorRecvLen = 0;
  memset(m_SensorBuffer, 0, sizeof(m_SensorBuffer));
  m_PlcRecvLen = 0;
  memset(m_PlcBuffer, 0, sizeof(m_PlcBuffer));
  m_RawRecvLen = 0;
  memset(m_RawBuffer, 0, sizeof(m_RawBuffer));

  m_ListenFigFlag = false;
  m_ListenPlcFlag = false;
  m_ListenRawFlag = false;
  m_proto_manager.reset(new SensorProtoManager);
}

StoreTask::~StoreTask() {
  PauseTask();
  for (auto it = m_fileList.begin(); it != m_fileList.end();) {
      it->second->uploadFileImmediate();
      it = m_fileList.erase(it);
  }
}

void StoreTask::StartTask() {
  readJsonFile(DATA_ENABLE_CONFIG, m_data_enable_);

  if (m_ListenFigFlag == false) {
    m_ListenFigFlag = m_data_enable_["FIGDATA"];
    m_ListenFigThread.reset(new std::thread([this] { ListenFigDataFunc(); }));
  }

  if (m_ListenPlcFlag == false) {
    m_ListenPlcFlag = m_data_enable_["PLCDATA"];
    m_ListenPlcThread.reset(new std::thread([this] { ListenPlcDataFunc(); }));
  }

  if (m_ListenRawFlag == false) {
    m_ListenRawFlag = m_data_enable_["RAWDATA"];
    m_ListenRawThread.reset(new std::thread([this] { ListenRawDataFunc(); }));
  }
}

void StoreTask::PauseTask() {
  if (m_ListenFigFlag) {
    m_ListenFigFlag = false;
    m_ListenFigThread->detach();
  }

  if (m_ListenPlcFlag) {
    m_ListenPlcFlag = false;
    m_ListenPlcThread->detach();
  }

  if (m_ListenRawFlag) {
    m_ListenRawFlag = false;
    m_ListenRawThread->detach();
  }
}

void StoreTask::ListenPlcDataFunc() {
  std::string gwipaddr;
  int32_t nret = searchFunc(PLC_CFG_FILE, "gw_ip", gwipaddr);
  if (nret < 0) {
    Common_Log_e("parse gwipaddr fail");
    return;
  }
  std::string plcZmqAddr = "tcp://" + gwipaddr + ":55554";

  void *context = zmq_init(1);
  void *sub = zmq_socket(context, ZMQ_SUB);
  zmq_connect(sub, plcZmqAddr.c_str());
  // zmq_setsockopt(sub, ZMQ_SUBSCRIBE, "CRRCPLC/0/0", strlen("CRRCPLC/0/0"));
  zmq_setsockopt(sub, ZMQ_SUBSCRIBE, "", strlen(""));

  /* 2024-08-10 17:15:26 */
  std::string startTime;
  std::string endTime;
  if (m_task_info.startTime != "") {
    startTime = m_task_info.startTime.substr(0, 4) +
                m_task_info.startTime.substr(5, 2) +
                m_task_info.startTime.substr(8, 2) +
                m_task_info.startTime.substr(11, 2) +
                m_task_info.startTime.substr(14, 2) +
                m_task_info.startTime.substr(17, 2);
  }

  if (m_task_info.endTime != "") {
    endTime =
        m_task_info.endTime.substr(0, 4) + m_task_info.endTime.substr(5, 2) +
        m_task_info.endTime.substr(8, 2) + m_task_info.endTime.substr(11, 2) +
        m_task_info.endTime.substr(14, 2) + m_task_info.endTime.substr(17, 2);
  }

  Common_Log_e("ListenPlcDataFunc start zmqIp:%s\n", plcZmqAddr.c_str());
  while (m_ListenPlcFlag && ((startTime == "" && endTime == "") ||
                             (Common_get_sys_time() < endTime &&
                              Common_get_sys_time() >= startTime))) {
    memset(m_PlcBuffer, 0, sizeof(m_PlcBuffer));
    m_PlcRecvLen = 0;
    int32_t nret = zmq_recv(sub, m_PlcBuffer, MAX_PLC_BUFF_LEN, 0);
    if (nret < 0) {
      Common_Log_e("sensor zmq recv fail");
      usleep(20000);
      continue;
    }

    m_PlcRecvLen += nret;
    // Common_Log_e("m_PlcBuffer, m_PlcRecvLen:%d\n", m_PlcRecvLen);
    // for (int index = 0; index < m_PlcRecvLen; index++) {
    //   printf("%x ", m_PlcBuffer[index]);
    //   if (index % 20 == 0) {
    //     printf("\n");
    //   }
    // }

    // 解析recvbuf
    uint8_t leftPlcBuffer[MAX_PLC_BUFF_LEN] = {0};
    int32_t leftBufferLen = 0;
    int32_t ret = m_proto_manager->DeSerializePlcData(
        m_PlcBuffer, m_PlcRecvLen, leftPlcBuffer, leftBufferLen);
    if (ret < 0) {
      Common_Log_e("proto DeSerialize error!!!");
      usleep(20000);
      continue;
    }
    // memset(m_PlcBuffer, 0, sizeof(m_PlcBuffer));
    // memcpy(m_PlcBuffer, leftPlcBuffer, leftBufferLen);

    usleep(20000);
  }
  m_ListenPlcThread->detach();
}

void StoreTask::ListenRawDataFunc() {
  void *context = zmq_ctx_new();
  void *sub = zmq_socket(context, ZMQ_SUB);
  std::string zmqurl = "tcp://" + m_task_info.zmqIp + ":55552";
  // std::string zmqurl = "tcp://192.168.31.108:55552";
  int rc = zmq_connect(sub, zmqurl.c_str());
  if (rc != 0) {
    Common_Log_e("zmq_connect fail!");
    return;
  }
  rc = zmq_setsockopt(sub, ZMQ_SUBSCRIBE, "", 0);
  if (rc != 0) {
    Common_Log_e("zmq_setsockopt fail!");
    return;
  }

  /* 2024-08-10 17:15:26 */
  std::string startTime;
  std::string endTime;
  if (m_task_info.startTime != "") {
    startTime = m_task_info.startTime.substr(0, 4) +
                m_task_info.startTime.substr(5, 2) +
                m_task_info.startTime.substr(8, 2) +
                m_task_info.startTime.substr(11, 2) +
                m_task_info.startTime.substr(14, 2) +
                m_task_info.startTime.substr(17, 2);
  }

  if (m_task_info.endTime != "") {
    endTime =
        m_task_info.endTime.substr(0, 4) + m_task_info.endTime.substr(5, 2) +
        m_task_info.endTime.substr(8, 2) + m_task_info.endTime.substr(11, 2) +
        m_task_info.endTime.substr(14, 2) + m_task_info.endTime.substr(17, 2);
  }

  Common_Log_e("ListenRawDataFunc start zmqurl:%s\n", zmqurl.c_str());
  while (m_ListenRawFlag && ((startTime == "" && endTime == "") ||
                             (Common_get_sys_time() < endTime &&
                              Common_get_sys_time() >= startTime))) {
    zmq_msg_t msg;
    bzero(&msg, sizeof(msg));
    zmq_msg_init(&msg);
    zmq_msg_recv(&msg, sub, 0); // 1非阻塞，0阻塞
    void *message = zmq_msg_data((zmq_msg_t *)&msg);
    size_t size = zmq_msg_size(&msg);

    if (size > MAX_RAW_BUFF_LEN) {
      Common_Log_e("raw data size:%d is too large\n", size);
      continue;
    }
    memset(m_RawBuffer, 0, sizeof(m_RawBuffer));
    m_RawRecvLen = 0;
    memcpy(m_RawBuffer + m_RawRecvLen, message, size);
    m_RawRecvLen += size;

    // 解析recvbuf
    uint8_t leftRawBuffer[MAX_RAW_BUFF_LEN] = {0};
    int32_t leftBufferLen = 0;
    std::string sensorKey;
    int32_t ret = m_proto_manager->DeSerializeRawData(
        m_RawBuffer, m_RawRecvLen, leftRawBuffer, leftBufferLen, m_task_info.mccAddress, sensorKey);
    if (ret < 0) {
      Common_Log_e("proto DeSerialize error!!!");
      usleep(50000);
      continue;
    }
    // memset(m_RawBuffer, 0, sizeof(m_RawBuffer));
    // memcpy(m_RawBuffer, leftRawBuffer, leftBufferLen);

    SensorData objData;
    m_proto_manager->GetSensorData(sensorKey, objData);
    uint16_t chanparam = objData.chanparam;

    int rawSize = objData.rawDataList.size();
    int figsize = objData.figureDataList.size();
    int plcsize = objData.plcDataList.size();
    // Common_Log_e("rawSize:%d figsize:%d plcsize:%d\n", rawSize, figsize,
    //              plcsize);
    for (int32_t index = 0; index < rawSize; index++) {
      SensorDataInfo::CommonSensor sensorData;
      memset(&sensorData, 0, sizeof(sensorData));
      sensorData.timestamp = objData.timems;
      sensorData.devChan = objData.sensorChan;
      sensorData.cpuID = objData.deviceId;
      sensorData.mccAddr = objData.addr;

      std::map<int32_t, rawData> rawItem = objData.rawDataList[index];

      int32_t itemIndex = 0;
      for (int32_t typeIndex = SensorDataInfo::CHAN_ACCX;
           typeIndex <= SensorDataInfo::CHAN_ACCZ; typeIndex++) {
        if (chanparam & (0x1 << typeIndex)) {
          figureData figdata = {0};
          rawData rawdata = rawItem[itemIndex];
          switch (typeIndex) {
          case SensorDataInfo::CHAN_ACCX: {
            AssignSensorData(figdata, rawdata, sensorData.accX);
            break;
          }
          case SensorDataInfo::CHAN_ACCY: {
            AssignSensorData(figdata, rawdata, sensorData.accY);
            break;
          }
          case SensorDataInfo::CHAN_ACCZ: {
            AssignSensorData(figdata, rawdata, sensorData.accZ);
            break;
          }
          }

          itemIndex++;
        }
      }

      //   printf("accX:%lf, accY:%lf, accZ:%lf\n", sensorData.accX.rawValue,
      //   sensorData.accY.rawValue, sensorData.accZ.rawValue);
      // 根据sn确定文件进行保存
      std::stringstream sensorType;
      sensorType << std::hex << objData.sn;
      sensorData.recvtime = Common_get_sys_time_msec() / 1000;
      std::string calendartime = Common_convert_utcms_to_calendar(sensorData.timestamp);
      memcpy(sensorData.calendartime, calendartime.c_str(), calendartime.length());
      std::string hourtime = Common_convert_utcms_to_hourtime(sensorData.timestamp);
      memcpy(sensorData.hourtime, hourtime.c_str(), hourtime.length());
      addSensorData(sensorData, sensorType.str());
    }

    usleep(50000);
  }

  m_ListenRawThread->detach();
}

void StoreTask::ListenFigDataFunc() {
  void *context = zmq_init(1);
  void *sub = zmq_socket(context, ZMQ_SUB);
  std::string zmqurl = "tcp://" + m_task_info.zmqIp + ":55553";
  zmq_connect(sub, zmqurl.c_str());
  zmq_setsockopt(sub, ZMQ_SUBSCRIBE, "", 0);
  // zmq_setsockopt(sub, ZMQ_SUBSCRIBE, "CRRCWS", 6);
  // zmq_setsockopt(sub, ZMQ_SUBSCRIBE, "CRRCCMO", 7);

  /* 2024-08-10 17:15:26 */
  std::string startTime;
  std::string endTime;
  if (m_task_info.startTime != "") {
    startTime = m_task_info.startTime.substr(0, 4) +
                m_task_info.startTime.substr(5, 2) +
                m_task_info.startTime.substr(8, 2) +
                m_task_info.startTime.substr(11, 2) +
                m_task_info.startTime.substr(14, 2) +
                m_task_info.startTime.substr(17, 2);
  }

  if (m_task_info.endTime != "") {
    endTime =
        m_task_info.endTime.substr(0, 4) + m_task_info.endTime.substr(5, 2) +
        m_task_info.endTime.substr(8, 2) + m_task_info.endTime.substr(11, 2) +
        m_task_info.endTime.substr(14, 2) + m_task_info.endTime.substr(17, 2);
  }

  Common_Log_e("ListenFigDataFunc start zmqurl:%s\n", zmqurl.c_str());
  while (m_ListenFigFlag && ((startTime == "" && endTime == "") ||
                             (Common_get_sys_time() < endTime &&
                              Common_get_sys_time() >= startTime))) {
    memset(m_SensorBuffer, 0, sizeof(m_SensorBuffer));
    m_SensorRecvLen = 0;
    int32_t nret = zmq_recv(sub, m_SensorBuffer, MAX_FIG_BUFF_LEN, 0);
    if (nret < 0) {
      Common_Log_e("sensor zmq recv fail nret:%d", nret);
      usleep(5000);
      continue;
    }

    if (nret > MAX_FIG_BUFF_LEN) {
      m_SensorRecvLen += MAX_FIG_BUFF_LEN;
    } else {
      m_SensorRecvLen += nret;
    }

    // printf("recvbuf: nret:%d\n", nret);
    // for (int index = 0; index < nret; index++) {
    //   printf("%x ", recvbuf[index]);
    //   if (index % 20 == 0) {
    //     printf("\n");
    //   }
    // }

    // 解析recvbuf
    uint8_t leftSensorBuffer[MAX_FIG_BUFF_LEN] = {0};
    int32_t leftBufferLen = 0;
    std::string sensorKey;
    int32_t ret = m_proto_manager->DeSerializeFigureData(
        m_SensorBuffer, m_SensorRecvLen, leftSensorBuffer, leftBufferLen, m_task_info.mccAddress,
        sensorKey);
    if (ret < 0) {
      usleep(5000);
      continue;
    }
    // memset(m_SensorBuffer, 0, sizeof(m_SensorBuffer));
    // memcpy(m_SensorBuffer, leftSensorBuffer, leftBufferLen);

    SensorData objData;
    ret = m_proto_manager->GetSensorData(sensorKey, objData);
    if (ret < 0) {
      continue;
    }

    // int circleSize = objData.figureDataList.size() >=
    // objData.rawDataList.size() ?
    //     objData.rawDataList.size() : objData.figureDataList.size();
    int32_t circleSize = 1;

    for (int32_t index = 0; index < circleSize; index++) {
      SensorDataInfo::CommonSensor sensorData;
      memset(&sensorData, 0, sizeof(sensorData));
      sensorData.timestamp = objData.timems;
      sensorData.devChan = objData.sensorChan;
      sensorData.cpuID = objData.deviceId;
      sensorData.mccAddr = objData.addr;
      if (objData.plcDataList.size() > 0) {
        sensorData.rotationSpeed = objData.plcDataList[0][0].GenSpeed;
        sensorData.windSpeed = objData.plcDataList[0][0].winspeed1;
        sensorData.power = objData.plcDataList[0][0].gridRealPower;
        sensorData.yaw = objData.plcDataList[0][0].yawstatecode;
      }

      std::map<int32_t, figureData> figItem = objData.figureDataList[index];
      std::map<int32_t, rawData> rawItem = objData.rawDataList[index];

      int32_t itemSize = figItem.size() > rawItem.size() ? rawItem.size() : figItem.size();
      auto figMapItem = figItem.begin();
      auto rawMapItem = rawItem.begin();
      for (int32_t itemIndex = 0; itemIndex < itemSize; itemIndex ++) {
        int32_t typeIndex = figMapItem->first;
        figureData figdata = figMapItem->second;
        rawData rawdata = rawMapItem->second;
        switch (typeIndex) {
        case SensorDataInfo::CHAN_ACCX: {
          AssignSensorData(figdata, rawdata, sensorData.accX);
          break;
        }
        case SensorDataInfo::CHAN_ACCY: {
          AssignSensorData(figdata, rawdata, sensorData.accY);
          break;
        }
        case SensorDataInfo::CHAN_ACCZ: {
          AssignSensorData(figdata, rawdata, sensorData.accZ);
          break;
        }
        case SensorDataInfo::CHAN_SPEEDX: {
          AssignSensorData(figdata, rawdata, sensorData.speedX);
          break;
        }
        case SensorDataInfo::CHAN_SPEEDY: {
          AssignSensorData(figdata, rawdata, sensorData.speedY);
          break;
        }
        case SensorDataInfo::CHAN_SPEEDZ: {
          AssignSensorData(figdata, rawdata, sensorData.speedZ);
          break;
        }
        case SensorDataInfo::CHAN_MOVEX: {
          AssignSensorData(figdata, rawdata, sensorData.moveX);
          break;
        }
        case SensorDataInfo::CHAN_MOVEY: {
          AssignSensorData(figdata, rawdata, sensorData.moveY);
          break;
        }
        case SensorDataInfo::CHAN_MOVEZ: {
          AssignSensorData(figdata, rawdata, sensorData.moveZ);
          break;
        }
        }

        figMapItem++;
        rawMapItem++;
      }

      // 根据sn确定文件进行保存
      std::stringstream sensorType;
      sensorType << std::hex << objData.sn;
      sensorData.recvtime = Common_get_sys_time_msec() / 1000;
      std::string calendartime = Common_convert_utcms_to_calendar(sensorData.timestamp);
      memcpy(sensorData.calendartime, calendartime.c_str(), calendartime.length());
      std::string hourtime = Common_convert_utcms_to_hourtime(sensorData.timestamp);
      memcpy(sensorData.hourtime, hourtime.c_str(), hourtime.length());
      addSensorData(sensorData, sensorType.str());
    }

    usleep(5000);
  }

  zmq_close(sub);
  zmq_term(context);

  if (Common_get_sys_time() > endTime) {
    for (auto it = m_fileList.begin(); it != m_fileList.end();) {
      it->second->uploadFileImmediate();
      it = m_fileList.erase(it);
    }
  }

  m_ListenFigThread->detach();
}

void StoreTask::AssignSensorData(const figureData &figure, const rawData &raw,
                                 SensorDataInfo::figureInfo &sensorInfo) {
  sensorInfo.rawValue = raw.rawValue;
  sensorInfo.absAverageValue = figure.absAverageValue;
  sensorInfo.averageValue = figure.averageValue;
  sensorInfo.kurtosis = figure.kurtosis;
  sensorInfo.mainAmp = figure.mainAmp;
  sensorInfo.mainFrequency = figure.mainFrequency;
  sensorInfo.margin = figure.margin;
  sensorInfo.maxValue = figure.maxValue;
  sensorInfo.minValue = figure.minValue;
  sensorInfo.peakFactor = figure.peakFactor;
  sensorInfo.peakValue = figure.peakValue;
  sensorInfo.pulseFactor = figure.pulseFactor;
  sensorInfo.standardDeviation = figure.standardDeviation;
  sensorInfo.variance = figure.variance;
  sensorInfo.waveFactor = figure.waveFactor;


  return;
}

void StoreTask::addSensorData(const SensorDataInfo::CommonSensor &sensorData,
                              const std::string &sensorType) {
  if (m_fileList.find(sensorType) == m_fileList.end()) {
    addFile(sensorType);
  }

  auto file = m_fileList[sensorType];
  file->addData(sensorData);
}

void StoreTask::addFile(const std::string &sensorType) {
  auto newFile = std::make_shared<StoreFile>();
  store_file_info fileInfo;
  fileInfo.coreCpuid = m_task_info.coreCpuID;
  fileInfo.storeCpuID = m_task_info.storeCpuID;
  fileInfo.file_dir = m_task_info.mccAddress + "/" + m_task_info.storeTaskName;
  fileInfo.mccaddr = m_task_info.mccAddress;
  fileInfo.sensor_name = sensorType;
  fileInfo.wsUrl = m_task_info.wsUrl;
  fileInfo.minioAccessKey = m_task_info.minioAccessKey;
  fileInfo.minioSecretKey = m_task_info.minioSecretKey;
  fileInfo.minioBucketName = m_task_info.minioBucketName;
  fileInfo.minioEndPoint = m_task_info.minioEndPoint;
  fileInfo.upload_data = m_task_info.upload_data;
  fileInfo.upload_file = m_task_info.upload_file;
  fileInfo.jsonInfo = m_task_info.jsonInfo;
  newFile->init(fileInfo);
  m_fileList[sensorType] = newFile;
}

void StoreTask::readJsonFile(const std::string &filename, json &objjson) {
    std::fstream file(filename);
    if (!file.is_open()) {
        Common_Log_e("filename:%s open fail", filename.c_str());
        return;
    }
    file >> objjson;

    return;
}
